#include "transmit_server.h"

#include <ctime>

#include "base.pb.h"
#include "chat_session_member.hxx"
#include "log.h"
#include "user.pb.h"
#include "uuid.h"

namespace chat {
// MsgTransmitServiceImpl

void MsgTransmitServiceImpl::GetTransmitTarget(
    ::google::protobuf::RpcController* controller,
    const ::chat::NewMessageReq* request,
    ::chat::GetTransmitTargetRsp* response, ::google::protobuf::Closure* done) {
    brpc::ClosureGuard closure_guard(done);

    std::string request_id = request->request_id();
    bool is_success = true;
    std::string err_msg = "OK";

    LOG_DEBUG("收到消息转发请求: {}", request_id);

    do {
        // 1. 从请求中取出消息内容，会话ID， 用户ID
        const std::string& user_id = request->user_id();
        const std::string& chat_session_id = request->chat_session_id();
        const MessageContent& message_content = request->message();

        // 2. 根据用户ID从用户子服务获取当前发送者用户信息
        chat::ServiceChannel::ChannelPtr channel =
            service_manager_->choose(user_service_name_);
        if (nullptr == channel) {
            LOG_ERROR("{}: 未找到能够提供服务 {} 的节点", request_id,
                      user_service_name_);

            is_success = false;
            err_msg = "未找到能够提供服务的节点";

            break;
        }
        chat::UserService_Stub stub(channel.get());
        chat::GetUserInfoReq req;
        chat::GetUserInfoRsp rsp;
        req.set_request_id(Uuid::uuid());
        req.set_user_id(user_id);
        req.set_session_id(request->session_id());

        brpc::Controller cntl;
        stub.GetUserInfo(&cntl, &req, &rsp, nullptr);
        if (cntl.Failed()) {
            LOG_ERROR("{}: {} 获取用户信息失败", request_id, user_id);

            is_success = false;
            err_msg = "获取用户信息失败";

            break;
        }
        if (!rsp.success()) {
            LOG_ERROR("{}: {} 获取用户信息响应错误", request_id, user_id);

            is_success = false;
            err_msg = "获取用户信息相应错误";

            break;
        }

        // 3. 根据消息内容构造完成的消息结构
        MessageInfo message_info;
        message_info.set_message_id(chat::Uuid::uuid());
        message_info.set_chat_session_id(chat_session_id);
        message_info.set_timestamp(time(nullptr) + 28800);
        message_info.mutable_sender()->CopyFrom(rsp.user_info());
        message_info.mutable_message()->CopyFrom(message_content);

        response->mutable_message()->CopyFrom(message_info);

        // 4.将消息发布到MQ消息队列中，让消息存储子服务进行持久化存储
        if (!mq_client_->publish(exchange_name_, routing_key_,
                                 message_info.SerializeAsString())) {
            LOG_ERROR("{}: 消息持久化失败", request_id);

            is_success = false;
            err_msg = "消息持久化失败";

            break;
        }

        // 5. 从数据库获取目标会话所有成员ID
        std::vector<std::string> member_ids =
            session_member_table_->members(chat_session_id);
        for (auto& id : member_ids) {
            response->add_target_id_list(id);
        }

    } while (0);

    // 6.组织响应，发送给网关，告知网关该将消息发送给谁
    response->set_request_id(request_id);
    response->set_success(is_success);
    response->set_errmsg(err_msg);

    LOG_DEBUG("成功处理消息转发请求: {}", request_id);
}

// -------------------------------------------------------------------

// MsgTransmitServerBuilder

void MsgTransmitServerBuilder::makeRegisterClient(
    const std::string& register_name, const std::string& etcd_host,
    const std::string& access_host) {
    register_client_ = std::make_shared<Registry>(etcd_host);

    register_client_->registry(register_name, access_host);
}

void MsgTransmitServerBuilder::makeServiceManager(
    const std::string& user_service_name) {
    service_manager_ = std::make_shared<ServiceManager>();
    service_manager_->declared(user_service_name);  // 添加关心

    user_service_name_ = user_service_name;
}

void MsgTransmitServerBuilder::makeDiscoveryClient(
    const std::string& etcd_host, const std::string& base_service) {
    discovery_client_ = std::make_shared<Discovery>(
        etcd_host, base_service,
        std::bind(&ServiceManager::onServiceOnline, service_manager_.get(),
                  std::placeholders::_1, std::placeholders::_2),
        std::bind(&ServiceManager::onServiceOffline, service_manager_.get(),
                  std::placeholders::_1, std::placeholders::_2));
}

void MsgTransmitServerBuilder::makeBrpcServer(const uint16_t port,
                                              const int idle_timeout_sec,
                                              const int num_threads) {
    // 检查
    if (odb_client_ == nullptr) {
        LOG_ERROR("ODB 客户端未初始化");
        abort();
    }
    if (mq_client_ == nullptr) {
        LOG_ERROR("MQ 客户端未初始化");
        abort();
    }
    if (service_manager_ == nullptr) {
        LOG_ERROR("服务管理对象 未初始化");
        abort();
    }

    brpc_server_ = std::make_shared<brpc::Server>();
    MsgTransmitServiceImpl* transmit_service = new MsgTransmitServiceImpl(
        service_manager_, user_service_name_, mq_client_, exchange_name_,
        routing_key_, odb_client_);
    if (-1 ==
        brpc_server_->AddService(transmit_service,
                                 brpc::ServiceOwnership::SERVER_OWNS_SERVICE)) {
        LOG_ERROR("RPC添加服务失败");
        abort();
    }

    brpc::ServerOptions options;
    options.idle_timeout_sec = idle_timeout_sec;
    options.num_threads = num_threads;

    if (-1 == brpc_server_->Start(port, &options)) {
        LOG_ERROR("RPC服务启动失败");
        abort();
    }
}

void MsgTransmitServerBuilder::makeOdbClient(
    const int max_pool, const std::string& user, const std::string& password,
    const std::string& db, const std::string& ip, const uint16_t port,
    const std::string& charset) {
    odb_client_ = DataBaseFactory::create(max_pool, user, password, db, ip,
                                          port, charset);
}

void MsgTransmitServerBuilder::makeMQClient(const std::string& user,
                                            const std::string& password,
                                            const std::string& host,
                                            const std::string& exchange_name,
                                            const std::string& queue_name,
                                            const std::string& routing_key) {
    exchange_name_ = exchange_name;
    routing_key_ = routing_key;
    mq_client_ = std::make_shared<MQClient>(user, password, host);
    mq_client_->declareCompoents(exchange_name, queue_name,
                                 AMQP::ExchangeType::direct, routing_key);
}

MsgTransmitServerPtr MsgTransmitServerBuilder::build() {
    if (register_client_ == nullptr) {
        LOG_FATAL("服务注册客户端未实例化");
        abort();
    }
    if (discovery_client_ == nullptr) {
        LOG_FATAL("服务发现客户端未实例化");
        abort();
    }
    if (brpc_server_ == nullptr) {
        LOG_FATAL("RPC服务未实例化");
        abort();
    }

    return std::make_shared<MsgTransmitServer>(
        register_client_, discovery_client_, brpc_server_, odb_client_);
}
}  // namespace chat